package com.zengqingfa.rocketmq.transaction;

import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 *
 * @fileName: TransactionProducer
 * @author: zengqf3
 * @date: 2021-4-2 16:21
 * @description:
 */
public class TransactionProducer {
    public static void main(String[] args) throws Exception {

        // 1、创建TransactionMQProducer
        TransactionMQProducer producer = new TransactionMQProducer("demo_producer_transaction_group");

        // 2、设置Namesrv地址
        producer.setNamesrvAddr("localhost:9876");

        // 3、指定消息监听对象，用于执行本地事务和消息回查
        producer.setTransactionListener(new TransactionListenerImpl());

        // 4、指定线程池
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2000), r -> {
            Thread thread = new Thread(r);
            thread.setName("client-transaction-msg-check-thread");
            return thread;
        });
        producer.setExecutorService(executorService);

        // 5、开启TransactionMQProducer
        producer.start();

        // 6、创建消息Message
        // topic：主题，tags: 标签，主要用于消息过滤，keys：消息的唯一值，body：消息体
        Message message = new Message("Topic_Transaction_Demo", "Tags", "Keys_T",
                "Hello Transaction RocketMQ Message".getBytes(RemotingHelper.DEFAULT_CHARSET));

        // 7、发送事务消息
        // 第一个参数：发送的消息信息
        // 第二个参数：选择指定的消息队列对象(会将所有的消息队列传进来)

        TransactionSendResult result = producer.sendMessageInTransaction(message, "hello-transaction");
        System.out.println(result);

        // 8、关闭TransactionMQProducer
        producer.shutdown();
    }
}
